package com.bw.dws;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsCarBehavior {
    public static void main(String[] args) {

        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
        evn.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(evn);


        tEnv.executeSql("CREATE TABLE dwd_cars_log (\n" +
                " id STRING,\n" +
                // 成员id
                " owerId INT,\n" +
                " opTime STRING,\n" +
                // 小区id
                " cId INT,\n" +
                // 车牌号
                " carCode STRING,\n" +
                " ctype INT,\n" +

                // 水位线
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(opTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '1' SECOND" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_cars_log',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'dwd_cars_log',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")"
        );


        // 窗口计算
        TableResult window = tEnv.executeSql(
                "select  " +
                        "TUMBLE_START(time_ltz,INTERVAL '1' second) as stt," +
                        "TUMBLE_END(time_ltz,INTERVAL '1' second) as edt," +

                        // 小区ID
                        "cId," +
                        // 业主id
                        "owerId," +
                        // 车牌号
                        "carCode," +

                        "  SUM(CASE WHEN ctype = 1 THEN 1 ELSE 0 END) AS in_cnt,\n" +
                        "  SUM(CASE WHEN ctype = 2 THEN 1 ELSE 0 END) AS out_cnt\n" +

                        " from dwd_cars_log " +
                        "WHERE cId IS NOT  null \n" +
                        "GROUP BY cId,owerId,carCode, TUMBLE(time_ltz,INTERVAL '1' second) "
        );

        window.print();

        tEnv.createTemporaryView("windowTable", (Table) window);


        tEnv.executeSql(
                "CREATE TABLE dws_car_behavior (\n" +

                        "stt VARCHAR(225) COMMENT '窗口开始日期时间' ,\n" +
                        "edt VARCHAR(225) COMMENT '窗口结束日期时间' ,\n" +

                        "cId INT COMMENT '小区ID' ,\n" +
                        "owerId INT COMMENT '业主id' ,\n" +
                        "carCode VARCHAR(225) COMMENT '车牌号' ,\n" +

                        "in_cnt BIGINT SUM COMMENT '驶入次数' ,\n" +
                        "out_cnt BIGINT SUM COMMENT '输出次数'  \n" +

                        ")WITH (\n" +
                        " 'connector' = 'doris',\n" +
                        " 'fenodes' = 'node102:8030',\n" +
                        " 'table.identifier' = 'sca_doris.dws_car_behavior',\n" +
                        " 'username' = 'root',\n" +
                        " 'password' = '123456',\n" +
                        " 'sink.label-prefix' = 'doris_label'\n" +
                        ");"
        );


        tEnv.executeSql("insert into dws_car_behavior \n" +
                "select \n" +
                "CAST(stt AS VARCHAR(225)) as stt , \n" +
                "CAST(edt AS VARCHAR(225)) as edt , \n" +

                "cId, \n" +
                "owerId, \n" +
                "carCode, \n" +

                "in_cnt, \n" +
                "out_cnt \n" +
                " from windowTable");



    }
}
